/**   
 * @Title: QueueListenter.java 
 * @Package com.cyl.rabbitmq 
 * @Description: TODO(��һ�仰�������ļ���ʲô) 
 * @author zjhua@hundsun.com   
 * @date 2016��4��25�� ����1:15:31 
 * @version V1.0   
 */
package com.medsoft.dlcache.mq;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.medsoft.base.utils.JsonUtils;
import com.medsoft.dlcache.CacheEntry;
import com.medsoft.dlcache.LocalCacheUpdateListener;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * @author zjhua
 *
 */
@Service
public class LocalCacheQueueListener implements InitializingBean {
	protected static final Logger log = LoggerFactory
			.getLogger(LocalCacheQueueListener.class);
	
	public static final String CACHE_OPERATION_REPLACE = "R";
	public static final String CACHE_OPERATION_DELETE = "D";
	
	private static Map<String,Class<CacheEntry>> clzMap = new HashMap<String,Class<CacheEntry>>();
	
	@Autowired
	private ConnectionFactory connectionFactory;

	@Value("${localcache.subscribeExchanges}")
	private String subscribeExchanges;
	
	private static Map<String,Channel> channelMap = new HashMap<String,Channel>();

	@Autowired
	private LocalCacheUpdateListener localCacheUpdateListener;
	
	/*
	 * (non-Javadoc)
	 * 
	 * 元数据格式：[OP][cacheName][S][@][cacheKey|cacheEntryStr|cacheEntriesStr]，其中OP长度为1，取值为R或D，cacheName长度不限，cacheKey|cacheEntryStr|cacheEntriesStr长度不限。cacheName后面如果跟了S，则代表替换整个缓存，而非某个具体的子项，仅用于数据字典子项。
	 */
	public void onMessage(String msg) {
		String updateCache = msg;
		String[] cacheFullKey = updateCache.substring(1).split("@");
		String cacheName = cacheFullKey[0];
		boolean isCaches = cacheName.substring(cacheName.length() - 1).equals("S");
		
		cacheName = (isCaches == true ? cacheName.substring(0, cacheName.length() - 2) : cacheName); //真正的缓存名称，不含@后缀
		String cacheKey = cacheFullKey[1];
		if (updateCache.substring(0, 1).equals(CACHE_OPERATION_DELETE)) { // 删除
			if (isCaches) {
				// NOP
				throw new UnsupportedOperationException();
			} else {
				localCacheUpdateListener.removeCacheEntry(cacheName, cacheKey);
			}
		} else if (updateCache.substring(0, 1).equals(CACHE_OPERATION_REPLACE)) { // 更新
			List<CacheEntry> cacheEntries;
			CacheEntry cacheEntry = null;
			ConcurrentHashMap<String, CacheEntry> cache = new ConcurrentHashMap<String, CacheEntry>();
			if (!clzMap.containsKey(cacheName)) {
				try {
					clzMap.put(cacheName,(Class<CacheEntry>) Thread.currentThread().getContextClassLoader().loadClass(cacheName));
				} catch (ClassNotFoundException e) {
					e.printStackTrace();
					throw new UnsupportedOperationException("找不到类" + cacheName + "，请确认缓存名和实际的POJO类名相同！");
				}
			}
			if(isCaches) {
				cacheEntries = JsonUtils.json2List(cacheKey,clzMap.get(cacheName));
				for (CacheEntry tmpCacheEntry : cacheEntries) {
					cache.put(tmpCacheEntry.getKey(), tmpCacheEntry);
				}
				localCacheUpdateListener.replaceCache(cacheName, cache);
			} else {
				cacheEntry = JsonUtils.deserialize(cacheKey,clzMap.get(cacheName));
				localCacheUpdateListener.saveCacheEntry(cacheName, cacheEntry);
			}
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
	 */
	@Override
	public void afterPropertiesSet() throws Exception {
		if(subscribeExchanges == null || subscribeExchanges.trim().equals(""))
		{
			System.out.println("本地缓存订阅队列参数[localcache.subscribeExchanges]没有设置，请检查server.properties");
			return;
		}
		Connection connection = connectionFactory.newConnection();

		for (String subscribeExchange : Arrays.asList(this.subscribeExchanges.split(","))) {
			Channel channel = connection.createChannel();
			channel.exchangeDeclare(subscribeExchange, "fanout");
			channelMap.put(subscribeExchange,channel);
			
			String queueName = channel.queueDeclare().getQueue();
			channel.queueBind(queueName,subscribeExchange, "");

			Consumer consumer = new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope,
						AMQP.BasicProperties properties, byte[] body)
						throws IOException {
					String message = new String(body, "UTF-8");
					log.info("收到本地缓存更新请求：[" + message + "]！");
					onMessage(message);
				}
			};
			channel.basicConsume(queueName, true, consumer);
		}
	}
}
